use std::collections::HashSet;
use std::ops::{Add, Sub};
use chrono::{DateTime, Local, TimeDelta};
use yew::UseReducerHandle;
use crate::db::pojo::pojo::{ImCollect, ImConfig, ImDepartment, ImFeedback, ImGroup, ImLabel, ImMessage, ImMessageActionStatus, ImModule, ImSensitiveWord, ImSession, ImSetting, ImUser};
use crate::db::service::im_collect_service::ImCollectService;
use crate::db::service::im_config_service::save_configs;
use crate::db::service::im_department_service::save_departments;
use crate::db::service::im_feedback_service::ImFeedbackService;
use crate::db::service::im_group_service::ImGroupService;
use crate::db::service::im_label_service::ImLabelService;
use crate::db::service::im_message_action_service::ImMessageActionService;
use crate::db::service::im_module_service::ImModuleService;
use crate::db::service::im_sensitive_word_service::ImSensitiveWordService;
use crate::db::service::im_session_service::ImSessionService;
use crate::db::service::im_setting_service::ImSettingService;
use crate::db::service::im_user_service::ImUserService;
use crate::business::app_state::AppState;
use crate::business::page::login_page::{LoadDataAction, LoadDataReducer};
use crate::business::rpc::service::rpc_group_service::RpcGroupService;
use crate::business::rpc::service::rpc_message_service::RpcMessageService;
use crate::business::rpc::service::rpc_pull_message_service::RpcPullMessageService;
use crate::business::service::message_send_service::MessageSendService;
use crate::business::service::save_message_service::SaveMessageService;
use crate::business::websocket::dto::im_base_dto::{CollectPullReq, ConfigPullReq, FeedbackPullReq, LabelPullReq, ModulePullReq, SensitiveWordPullReq};
use crate::business::websocket::dto::im_department_dto::DepartmentInfoListReq;
use crate::business::websocket::dto::im_group_dto::{GroupCreateReq, GroupInfoListReq, GroupType};
use crate::business::websocket::dto::im_message_dto::{MessageActionItem, MessageActionOkItem, MessageActionType, MessagePullActionOkReq, MessagePullActionReq, MessagePullBySessionIdsReq, MessageReadReq, MessageWithdrawReq, OkListReq};
use crate::business::websocket::dto::im_session_dto::{SessionPullReq, SessionType};
use crate::business::websocket::dto::user_dto::{TerminalType, UserInfoListReq};
use crate::db::dao::im_group_dao::get_need_send_groups_by_user_id;
use crate::db::dao::im_message_dao::{change_message_action_status2, query_need_send_messages_by_user_id};
use crate::db::db::Db;
use crate::util::time::{format_time_by_format, parse_to_time_by_format};

pub struct SynDataService;

impl SynDataService {

    pub async fn syn_data(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>) {
        log::info!("syn data started, user_id: {}", &user_id);
        let start_time = Local::now();

        self.syn_common_data(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;

        let end_time = Local::now();
        let time = end_time.sub(start_time).as_seconds_f32();
        log::info!("syn data total time: {}", time);
    }

    async fn syn_common_data(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        let mut total = 0;
        total += self.syn_configs(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.syn_departments(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.syn_users(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.syn_groups(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.syn_modules(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.send_need_groups(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.send_need_messages(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.send_message_action_status(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.syn_sessions(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.syn_messages(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.syn_message_actions(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.syn_labels(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.syn_collects(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.syn_feedbacks(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        total += self.syn_sensitive_words(app_state.clone(), user_id.clone(), load_data_reducer.clone()).await;
        log::info!("syn data total: {}", total);
        total
    }

    async fn syn_configs(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        let t = self.syn_data_wrap(&user_id.clone(), "SynConfig", async move |start_time, _is_last_change|->usize {
            let websocket_status = app_state.websocket_status.clone();
            let page_size = 100;

            log::info!("syn configs websocket_status: {:?}", websocket_status);

            let pull_message_service = RpcPullMessageService;
            let mut start_server_name = None;
            let mut total = 0;
            let mut loop_times = 0;

            loop {
                let config_pull_req = ConfigPullReq {
                    page_size,
                    start_date: start_time,
                    start_server_name: start_server_name.clone(),
                    create_date_time: Local::now(),
                    local_seq: "".to_string(),
                };

                let result = pull_message_service.pull_config_list(app_state.clone(), config_pull_req.clone()).await;
                match result {
                    Ok(config_item_list) => {
                        let mut im_configs = config_item_list.config_items.into_iter().map(|config_item| { config_item.into() }).collect::<Vec<ImConfig>>();
                        total += im_configs.len();
                        save_configs(user_id.clone(), &mut im_configs);

                        let is_end = im_configs.len() < page_size;
                        log::info!("im_configs pull is end: {:?}", is_end);
                        let lasted = im_configs.last();
                        start_server_name = match lasted {
                            None => {None}
                            Some(value) => {
                                Some(value.name.clone())
                            }
                        };

                        if is_end {
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(5.0, "同步完成配置信息...".to_string()));
                            break;
                        } else {
                            loop_times += 1;
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(0.5, format!("正在同步配置信息{}...", loop_times)));
                        }
                    }
                    Err(e) => {
                        log::error!("syn config pull error: {:?}", e);
                        break;
                    }
                }
            }
            total
        }).await;
        t
    }

    async fn syn_departments(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        self.syn_data_wrap(&user_id.clone(), "SynDepartment", async move |start_date, _is_last_change|->usize {
            let websocket_status = app_state.websocket_status.clone();
            let page_size = 100;

            log::info!("syn departments websocket_status: {:?}", websocket_status);

            let mut start_org_code_option = None;
            let mut total = 0;
            let mut loop_times = 0;

            let pull_message_service = RpcPullMessageService;

            loop {
                let start_org_code = start_org_code_option.clone();
                let department_info_list_req = DepartmentInfoListReq {
                    start_org_code,
                    page_size,
                    start_date,
                    is_latest_changed: true,
                    create_date_time: Local::now(),
                    local_seq: "".to_string(),
                };

                let result = pull_message_service.pull_department_list(app_state.clone(), department_info_list_req.clone()).await;
                match result {
                    Ok(department_info_item_list) => {
                        let mut im_departments = department_info_item_list.department_info_item_list.into_iter().map(|department_info| { department_info.into() }).collect::<Vec<ImDepartment>>();
                        total += im_departments.len();
                        save_departments(user_id.clone(), &mut im_departments);

                        let is_end = im_departments.len() < page_size;
                        log::info!("syn departments pull is end: {:?}", is_end);
                        let lasted = im_departments.last();
                        start_org_code_option = match lasted {
                            None => {None}
                            Some(value) => {
                                Some(value.code.clone())
                            }
                        };

                        if is_end {
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(5.0, "同步完成机构信息...".to_string()));
                            break;
                        } else {
                            loop_times += 1;
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(0.5, format!("正在同步配置信息{}...", loop_times)));
                        }
                    }
                    Err(e) => {
                        log::error!("syn departments pull error: {:?}", e);
                        break;
                    }
                }
            }
            total
        }).await
    }

    async fn syn_users(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        self.syn_data_wrap(&user_id.clone(), "SynUser", async move |start_date, _is_last_change|->usize {
            let websocket_status = app_state.websocket_status.clone();
            let page_size = 200;

            log::info!("syn users websocket_status: {:?}", websocket_status);

            let mut start_user_id_option = None;
            let mut total = 0;
            let mut loop_times = 0;

            let pull_message_service = RpcPullMessageService;
            let user_service = ImUserService;
            loop {
                let start_user_id = start_user_id_option.clone();
                let user_info_list_req = UserInfoListReq {
                    create_date_time: Local::now(),
                    local_seq: "".to_string(),
                    start_user_id,
                    page_size: page_size as u32,
                    start_date,
                    is_latest_change: true,
                };

                let result = pull_message_service.pull_user_list(app_state.clone(), user_info_list_req.clone()).await;
                match result {
                    Ok(user_info_item_list) => {
                        let mut im_users = user_info_item_list.user_info_item_list.into_iter().map(|user_info| { user_info.into() }).collect::<Vec<ImUser>>();
                        total += im_users.len();
                        user_service.save_users(&mut im_users);

                        let is_end = im_users.len() < page_size;
                        log::info!("syn users pull is end: {:?}", is_end);
                        let lasted = im_users.last();
                        start_user_id_option = match lasted {
                            None => {None}
                            Some(value) => {
                                Some(value.user_id.clone())
                            }
                        };

                        if is_end {
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(5.0, "同步完成用户信息...".to_string()));
                            break;
                        } else {
                            loop_times += 1;
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(1.0, format!("正在同步用户信息{}...", loop_times)));
                        }
                    }
                    Err(e) => {
                        log::error!("syn users pull error: {:?}", e);
                        break;
                    }
                }
            }
            total
        }).await
    }

    async fn syn_groups(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        self.syn_data_wrap(&user_id.clone(), "SynGroup", async move |start_date, _is_last_change|->usize {
            let websocket_status = app_state.websocket_status.clone();
            let page_size = 20;

            log::info!("syn groups websocket_status: {:?}", websocket_status);

            let mut start_server_id_option = None;
            let mut total = 0;
            let mut loop_times = 0;

            let pull_message_service = RpcPullMessageService;
            let group_service = ImGroupService;

            loop {
                let start_server_id = start_server_id_option.clone();
                let group_info_list_req = GroupInfoListReq {
                    group_type: GroupType::Chat,
                    create_date_time: Local::now(),
                    local_seq: "".to_string(),
                    page_size,
                    group_id: None,
                    start_date,
                    start_server_id,
                    session_id: None,
                };

                let result = pull_message_service.pull_group_list(app_state.clone(), group_info_list_req.clone()).await;
                match result {
                    Ok(group_info_item_list) => {
                        let mut im_groups = group_info_item_list.group_info_item_list.into_iter().map(|group_info| { group_info.into() }).collect::<Vec<ImGroup>>();
                        total += im_groups.len();
                        group_service.save_groups(&mut im_groups);

                        let is_end = im_groups.len() < page_size;
                        log::info!("syn groups pull is end: {:?}", is_end);
                        let lasted = im_groups.last();
                        start_server_id_option = match lasted {
                            None => {None}
                            Some(value) => {
                                Some(value.code.clone())
                            }
                        };

                        if is_end {
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(2.0, "同步完成群信息...".to_string()));
                            break;
                        } else {
                            loop_times += 1;
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(1.0, format!("正在同步群信息{}...", loop_times)));
                        }
                    }
                    Err(e) => {
                        log::error!("syn groups pull error: {:?}", e);
                        break;
                    }
                }
            }
            total
        }).await
    }

    async fn syn_modules(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        self.syn_data_wrap(&user_id.clone(), "SynModule", async move |start_date, _is_last_change|->usize {
            let websocket_status = app_state.websocket_status.clone();
            let page_size = 20;

            log::info!("syn modules websocket_status: {:?}", websocket_status);

            let mut start_server_name_option = None;
            let mut total = 0;
            let mut loop_times = 0;

            let pull_message_service = RpcPullMessageService;
            let module_service = ImModuleService;

            loop {
                let start_server_name = start_server_name_option.clone();
                let module_pull_req = ModulePullReq {
                    page_size,
                    start_date,
                    start_server_name,
                    create_date_time: Default::default(),
                    local_seq: "".to_string(),
                };

                let result = pull_message_service.pull_module_list(app_state.clone(), module_pull_req.clone()).await;
                match result {
                    Ok(module_info_item_list) => {
                        let mut im_modules = module_info_item_list.module_info_items.into_iter().map(|module_info| { module_info.into() }).collect::<Vec<ImModule>>();
                        total += im_modules.len();
                        module_service.save_modules(&mut im_modules);

                        let is_end = im_modules.len() < page_size;
                        log::info!("syn modules pull is end: {:?}", is_end);
                        let lasted = im_modules.last();
                        start_server_name_option = match lasted {
                            None => {None}
                            Some(value) => {
                                Some(value.name.clone())
                            }
                        };

                        if is_end {
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(2.0, "同步完成模块信息...".to_string()));
                            break;
                        } else {
                            loop_times += 1;
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(1.0, format!("正在同步模块信息{}...", loop_times)));
                        }
                    }
                    Err(e) => {
                        log::error!("syn modules pull error: {:?}", e);
                        break;
                    }
                }
            }
            total
        }).await
    }


   async fn send_need_groups(&self, app_state: AppState, user_id: String, _load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        let db = Db::open();
        let mut size = 0;
        let page_size = 1000;
        let groups = get_need_send_groups_by_user_id(db.clone(), user_id.clone(), page_size);
        size += groups.len();

        for mut group in groups.into_iter() {
            let group_create_req = GroupCreateReq {
                group_name: group.name.clone().unwrap(),
                group_desc: group.description.clone(),
                user_id_set: group.group_users.clone().unwrap().into_iter().map(|group_user| { group_user.user_id }).collect(),
                group_id: group.code.clone(),
                session_id: None,
                create_date_time: Local::now(),
                local_seq: "".to_string(),
            };

            let result = RpcGroupService::group_create(app_state.clone(), group_create_req).await;

            match result {
                Ok(ok_resp) => {
                    let session_server_id = ok_resp.other_info.unwrap();
                    ImGroupService::send_success(&mut group, session_server_id);
                }
                Err(_) => {}
            }
        }
        size
    }

    async fn send_need_messages(&self, app_state: AppState, user_id: String, _load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        let db = Db::open();
        let mut size = 0;
        size += self.send_common_messages(db.clone(), app_state.clone(), &user_id).await;
        size += self.send_file_messages(db.clone(), app_state.clone(), &user_id).await;
        size
    }

    async fn send_message_action_status(&self, app_state: AppState, user_id: String, _load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        let db = Db::open();
        let page_size = 500;
        let mut size = 0;
        loop {
            let messages = query_need_send_messages_by_user_id(db.clone(), &user_id, page_size);
            let length = messages.len();
            size += messages.len();

            let mut message_read_messages = vec![];
            let mut message_withdraw_messages = vec![];
            for message in messages.into_iter() {
                if message.action_status == ImMessageActionStatus::Read {
                    message_read_messages.push(message);
                } else if message.action_status == ImMessageActionStatus::Withdraw {
                    message_withdraw_messages.push(message);
                }
            }
            self.send_read_messages(db.clone(), app_state.clone(), &user_id, message_read_messages).await;
            self.send_withdraw_messages(db.clone(), app_state.clone(), &user_id, message_withdraw_messages).await;

            if length != page_size {
                break;
            }
        }
        size
    }

    async fn syn_sessions(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        self.syn_data_wrap(&user_id.clone(), "SynSession", async move |start_date, _is_last_change|->usize {
            let websocket_status = app_state.websocket_status.clone();
            let page_size = 100;

            log::info!("syn sessions websocket_status: {:?}", websocket_status);

            let mut start_server_id_option = None;
            let mut total = 0;
            let mut loop_times = 0;

            let pull_message_service = RpcPullMessageService;
            let session_service = ImSessionService;
            loop {
                let start_server_id = start_server_id_option.clone();
                let session_pull_req = SessionPullReq {
                    page_size,
                    start_date,
                    session_type: SessionType::Chat,
                    start_server_id,
                    create_date_time: Local::now(),
                    local_seq: "".to_string(),
                };

                let result = pull_message_service.pull_session_list(app_state.clone(), session_pull_req.clone()).await;
                match result {
                    Ok(session_item_list) => {
                        let mut im_sessions = session_item_list.session_item_list.into_iter()
                            .map(|session_item| ImSession::wrap(session_item, user_id.clone())).collect::<Vec<ImSession>>();

                        total += im_sessions.len();
                        session_service.save_sessions(&mut im_sessions);

                        let is_end = im_sessions.len() < page_size;
                        log::info!("syn sessions pull is end: {:?}", is_end);
                        let lasted = im_sessions.last();
                        start_server_id_option = match lasted {
                            None => {None}
                            Some(value) => {
                                value.server_id.clone()
                            }
                        };

                        if is_end {
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(2.0, "同步完成会话信息...".to_string()));
                            break;
                        } else {
                            loop_times += 1;
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(1.0, format!("正在同步会话信息{}...", loop_times)));
                        }
                    }
                    Err(e) => {
                        log::error!("syn sessions pull error: {:?}", e);
                        break;
                    }
                }
            }
            total
        }).await
    }

    async fn send_read_messages(&self, db: Db, app_state: AppState,  _user_id: &str, message_read_messages: Vec<ImMessage>) {
        if message_read_messages.is_empty() {
            return;
        }

        let (message_ids, message_server_ids): (Vec<u64>, Vec<String>) = message_read_messages.into_iter()
            .map(|message| (message.id.unwrap(), message.server_id.unwrap()))
            .unzip();
        let message_read_req = MessageReadReq {
            message_ids: message_server_ids.into_iter().collect::<HashSet<String>>(),
            create_date_time: Local::now(),
            local_seq: None,
        };
        let result = RpcMessageService::send_read_message(app_state.clone(), message_read_req).await;
        match result {
            Ok(_ok_resp) => {
                change_message_action_status2(db.clone(), message_ids.into_iter().collect(), ImMessageActionStatus::Readed);
            }
            Err(exception_resp) => {
                log::error!("send message read action error: {:?}", exception_resp);
            }
        }
    }

    async fn send_withdraw_messages(&self, db: Db, app_state: AppState, _user_id: &str, message_withdraw_messages: Vec<ImMessage>) {
        if message_withdraw_messages.is_empty() {
            return;
        }

        let (message_ids, message_server_ids): (Vec<u64>, Vec<String>) = message_withdraw_messages.into_iter()
            .map(|message| (message.id.unwrap(), message.server_id.unwrap()))
            .unzip();
        let message_withdraw_req = MessageWithdrawReq {
            message_ids: message_server_ids.into_iter().collect::<HashSet<String>>(),
            create_date_time: Local::now(),
            local_seq: None,
        };
        let result = RpcMessageService::send_withdraw_message(app_state.clone(), message_withdraw_req).await;
        match result {
            Ok(_ok_resp) => {
                change_message_action_status2(db.clone(), message_ids.into_iter().collect(), ImMessageActionStatus::Withdrawed);
            }
            Err(exception_resp) => {
                log::error!("send message withdraw action error: {:?}", exception_resp);
            }
        }
    }


    async fn syn_messages(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        let user_id_clone = user_id.clone();
        self.syn_data_wrap(&user_id.clone(), "SynMessage", async move |start_date, _is_last_change|->usize {
            let websocket_status = app_state.websocket_status.clone();
            let page_size = 500;

            log::info!("syn messages websocket_status: {:?}", websocket_status);

            let mut start_server_message_id_option = None;
            let mut total = 0;
            let mut loop_times = 0;

            let pull_message_service = RpcPullMessageService;
            let session_service = ImSessionService;

            let sessions = session_service.query(&user_id, 0, 500);
            let session_ids = sessions.into_iter().map(|session| session.server_id.unwrap()).collect::<HashSet<String>>();

            loop {
                let start_server_message_id = start_server_message_id_option.clone();
                let session_ids = session_ids.clone();
                let message_pull_req = MessagePullBySessionIdsReq {
                    start_server_message_id,
                    page_size,
                    session_ids,
                    terminal_type: TerminalType::Web,
                    aft_date: start_date,
                };

                let result = pull_message_service.pull_message_list(app_state.clone(), message_pull_req.clone()).await;
                match result {
                    Ok(message_object_list) => {
                        log::info!("syn messages pull is end: {:?}", message_object_list);
                        let local_seqs = message_object_list.message_object_list.iter().map(|message| { message.local_seq.clone() }).collect::<HashSet<String>>();
                        let message_objects = message_object_list.message_object_list;
                        total += message_objects.len();

                        let size = message_objects.len();

                        let is_end = size < page_size;
                        log::info!("syn messages pull is end: {:?}", is_end);
                        let lasted = message_objects.last();
                        start_server_message_id_option = match lasted {
                            None => {None}
                            Some(value) => {
                                value.server_id.clone()
                            }
                        };

                        SaveMessageService::save_messages(user_id_clone.clone(), message_objects,true);

                        if local_seqs.len() > 0 {
                            let ok_list_req = OkListReq {
                                local_seq_set: local_seqs,
                                coder: 11,
                                other_info: None,
                            };
                            let _ = pull_message_service.ok_messages(app_state.clone(), ok_list_req).await;
                        }

                        if is_end {
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(2.0, "同步完成消息信息...".to_string()));
                            break;
                        } else {
                            loop_times += 1;
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(1.0, format!("正在同步消息信息{}...", loop_times)));
                        }
                    }
                    Err(e) => {
                        log::error!("syn messages pull error: {:?}", e);
                        break;
                    }
                }
            }
            total
        }).await
    }

    async fn syn_message_actions(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        self.syn_data_wrap(&user_id.clone(), "SynMessageAction", async move |start_date, _is_last_change|->usize {
            let websocket_status = app_state.websocket_status.clone();
            let page_size = 20;

            log::info!("syn message actions websocket_status: {:?}", websocket_status);

            let mut server_message_id_option = None;
            let mut total = 0;

            let mut loop_index = 0;
            let message_action_service = ImMessageActionService;
            let pull_message_service = RpcPullMessageService;

            loop {
                let server_message_id = server_message_id_option.clone();
                let message_pull_req = MessagePullActionReq {
                    server_message_id,
                    size: page_size,
                    aft_date: start_date,
                    create_date_time: Local::now(),
                    local_seq: "".to_string(),
                };

                let message_pull_data_result = pull_message_service.pull_message_action_list(app_state.clone(), message_pull_req).await;
                match message_pull_data_result {
                    Ok(message_action_list) => {
                        let message_action_list = message_action_list.message_action_item_list;
                        total += message_action_list.len();

                        message_action_service.save_message_actions(user_id.clone(), &message_action_list);

                        let ok_action_items = message_action_list.iter().filter(|action_item_list|  action_item_list.r#type == MessageActionType::Owner).collect::<Vec<&MessageActionItem>>();
                        if ok_action_items.len() > 0 {
                            let message_action_ok_resp_list = ok_action_items.into_iter()
                                .map(|action_item| {action_item.into()}).collect::<Vec<MessageActionOkItem>>();

                            let message_pull_action_ok_req = MessagePullActionOkReq {
                                message_action_ok_item_list: message_action_ok_resp_list,
                                create_date_time: Local::now(),
                                local_seq: "".to_string(),
                            };

                            let _ok_message_action_result = pull_message_service.ok_message_actions(app_state.clone(), message_pull_action_ok_req).await;
                        }

                        let is_end = message_action_list.len() < page_size;
                        log::info!("syn message actions pull is end: {:?}", is_end);
                        let lasted = message_action_list.last();
                        server_message_id_option = match lasted {
                            None => {None}
                            Some(value) => {
                                Some(value.message_id.clone())
                            }
                        };

                        if is_end {
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(1.0, "同步完成消息事件...".to_string()));
                            break;
                        } else {
                            loop_index = loop_index + 1;
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(0.5, format!("正在同步消息事件{}...", loop_index)));
                        }
                    }
                    Err(e) => {
                        log::error!("syn message actions pull error: {:?}", e);
                        break;
                    }
                }
            }
            total
        }).await
    }

    async fn syn_labels(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        self.syn_data_wrap(&user_id.clone(), "SynLabel", async move |start_date, _is_last_change|->usize {
            let websocket_status = app_state.websocket_status.clone();
            let page_size = 200;

            log::info!("syn labels websocket_status: {:?}", websocket_status);

            let mut server_start_id_option = None;
            let mut total = 0;

            let mut loop_index = 0;
            let label_service = ImLabelService;
            let pull_message_service = RpcPullMessageService;

            loop {
                let server_start_id = server_start_id_option.clone();
                let label_pull_req = LabelPullReq {
                    page_size,
                    server_start_id,
                    start_date,
                };

                let label_pull_result = pull_message_service.pull_label_list(app_state.clone(), label_pull_req).await;
                match label_pull_result {
                    Ok(label_info_item_list) => {

                        let is_end = label_info_item_list.label_info_items.len() < page_size;
                        log::info!("syn collects pull is end: {:?}", is_end);
                        let lasted = label_info_item_list.label_info_items.last();
                        server_start_id_option = match lasted {
                            None => {None}
                            Some(value) => {
                                Some(value.server_id.clone())
                            }
                        };

                        let mut im_labels = label_info_item_list.label_info_items.into_iter().map(|label_info_item| {label_info_item.into()}).collect::<Vec<ImLabel>>();
                        total += im_labels.len();

                        label_service.save_labels(user_id.clone(), &mut im_labels);

                        if is_end {
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(2.0, "同步完成标签信息...".to_string()));
                            break;
                        } else {
                            loop_index = loop_index + 1;
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(0.5, format!("正在同步标签信息{}...", loop_index)));
                        }
                    }
                    Err(e) => {
                        log::error!("syn collects pull error: {:?}", e);
                        break;
                    }
                }
            }
            total
        }).await
    }


    async fn syn_collects(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        self.syn_data_wrap(&user_id.clone(), "SynCollect", async move |start_date, _is_last_change|->usize {
            let websocket_status = app_state.websocket_status.clone();
            let page_size = 200;

            log::info!("syn collects websocket_status: {:?}", websocket_status);

            let mut server_start_id_option = None;
            let mut total = 0;

            let mut loop_index = 0;
            let collect_service = ImCollectService;
            let pull_message_service = RpcPullMessageService;

            loop {
                let server_start_id = server_start_id_option.clone();
                let collect_pull_req = CollectPullReq {
                    page_size,
                    server_start_id,
                    start_date,
                    create_date_time: Local::now(),
                    local_seq: "".to_string(),
                };

                let collect_pull_result = pull_message_service.pull_collect_list(app_state.clone(), collect_pull_req).await;
                match collect_pull_result {
                    Ok(collect_item_list) => {

                        let is_end = collect_item_list.collect_info_items.len() < page_size;
                        log::info!("syn collects pull is end: {:?}", is_end);
                        let lasted = collect_item_list.collect_info_items.last();
                        server_start_id_option = match lasted {
                            None => {None}
                            Some(value) => {
                                Some(value.server_id.clone())
                            }
                        };

                        let mut im_collects = collect_item_list.collect_info_items.into_iter().map(|collect_info_item| {collect_info_item.into()}).collect::<Vec<ImCollect>>();
                        total += im_collects.len();

                        collect_service.save_collects(user_id.clone(), &mut im_collects);

                        if is_end {
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(2.0, "同步完成收藏信息...".to_string()));
                            break;
                        } else {
                            loop_index = loop_index + 1;
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(0.5, format!("正在同步收藏信息{}...", loop_index)));
                        }
                    }
                    Err(e) => {
                        log::error!("syn labels pull error: {:?}", e);
                        break;
                    }
                }
            }
            total
        }).await
    }

    async fn syn_feedbacks(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        self.syn_data_wrap(&user_id.clone(), "SynFeedback", async move |start_date, _is_last_change|->usize {
            let websocket_status = app_state.websocket_status.clone();
            let page_size = 200;

            log::info!("syn feedbacks websocket_status: {:?}", websocket_status);

            let mut server_start_id_option = None;
            let mut total = 0;

            let mut loop_index = 0;
            let feedback_service = ImFeedbackService;
            let pull_message_service = RpcPullMessageService;

            loop {
                let server_start_id = server_start_id_option.clone();
                let feedback_pull_req = FeedbackPullReq {
                    page_size,
                    server_start_id,
                    start_date,
                    create_date_time: Local::now(),
                    local_seq: "".to_string(),
                };

                let feedback_pull_result = pull_message_service.pull_feedback_list(app_state.clone(), feedback_pull_req).await;
                match feedback_pull_result {
                    Ok(feedback_item_list) => {

                        let is_end = feedback_item_list.feedback_item_list.len() < page_size;
                        log::info!("syn feedbacks pull is end: {:?}", is_end);
                        let lasted = feedback_item_list.feedback_item_list.last();
                        server_start_id_option = match lasted {
                            None => {None}
                            Some(value) => {
                                Some(value.server_id.clone())
                            }
                        };

                        let mut im_feedbacks = feedback_item_list.feedback_item_list.into_iter().map(|feedback_info_item| {feedback_info_item.into()}).collect::<Vec<ImFeedback>>();
                        total += im_feedbacks.len();

                        feedback_service.save_feedbacks(user_id.clone(), &mut im_feedbacks);

                        if is_end {
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(2.0, "同步完成反馈信息...".to_string()));
                            break;
                        } else {
                            loop_index = loop_index + 1;
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(1.0, format!("正在同步反馈信息{}...", loop_index)));
                        }
                    }
                    Err(e) => {
                        log::error!("syn feedbacks pull error: {:?}", e);
                        break;
                    }
                }
            }
            total
        }).await
    }


    async fn syn_sensitive_words(&self, app_state: AppState, user_id: String, load_data_reducer: UseReducerHandle<LoadDataReducer>)->usize {
        self.syn_data_wrap(&user_id.clone(), "SynSensitiveWord", async move |start_date, _is_last_change|->usize {
            let websocket_status = app_state.websocket_status.clone();
            let page_size = 200;

            log::info!("syn sensitive word websocket_status: {:?}", websocket_status);

            let mut server_start_id_option = None;
            let mut total = 0;

            let mut loop_index = 0;
            let sensitive_word_service = ImSensitiveWordService;
            let pull_message_service = RpcPullMessageService;

            loop {
                let server_start_id = server_start_id_option.clone();
                let sensitive_word_pull_req = SensitiveWordPullReq {
                    page_size,
                    server_start_id,
                    start_date,
                    create_date_time: Local::now(),
                    local_seq: "".to_string(),
                };

                let sensitive_word_pull_result = pull_message_service.pull_sensitive_word_list(app_state.clone(), sensitive_word_pull_req).await;
                match sensitive_word_pull_result {
                    Ok(sensitive_word_list) => {

                        let is_end = sensitive_word_list.sensitive_word_list.len() < page_size;
                        log::info!("syn sensitive word pull is end: {:?}", is_end);
                        let lasted = sensitive_word_list.sensitive_word_list.last();
                        server_start_id_option = match lasted {
                            None => {None}
                            Some(value) => {
                                value.server_id.clone()
                            }
                        };

                        let mut im_sensitive_words = sensitive_word_list.sensitive_word_list.into_iter().map(|sensitive_word_item| {sensitive_word_item.into()}).collect::<Vec<ImSensitiveWord>>();
                        total += im_sensitive_words.len();

                        sensitive_word_service.save_sensitive_words(user_id.clone(), &mut im_sensitive_words);

                        if is_end {
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(2.0, "同步完成敏感词信息...".to_string()));
                            load_data_reducer.dispatch(LoadDataAction::Done("同步完成敏感词信息...".to_string()));
                            break;
                        } else {
                            loop_index = loop_index + 1;
                            load_data_reducer.dispatch(LoadDataAction::SendProgressAndMsg(1.0, format!("正在同步敏感词信息{}...", loop_index)));
                        }
                    }
                    Err(e) => {
                        log::error!("syn sensitive word pull error: {:?}", e);
                        break;
                    }
                }
            }
            total
        }).await
    }

    async fn send_common_messages(&self, db: Db, app_state: AppState, user_id: &str) ->usize {
        let page_size = 300;
        let mut messages = query_need_send_messages_by_user_id(db.clone(), &user_id, page_size);
        let size = messages.len();

        log::info!("syn common messages: {}, {:?}", user_id, messages);

        if !messages.is_empty() {
            MessageSendService::send_messages(app_state.clone(), &mut messages).await;
        }

        size
    }

    async fn send_file_messages(&self, _db: Db, _app_state: AppState, _user_id: &str) ->usize {
        let size = 0;
        //todo

        size
    }


    async fn syn_data_wrap<F, Fut>(&self, user_id: &String, name: &'static str, do_business: F) ->usize
        where F: FnOnce(Option<DateTime<Local>>, bool) -> Fut + 'static,  // 闭包自身需静态存活
              Fut: Future<Output = usize> + 'static{
        let setting_service = ImSettingService;
        let db_setting = setting_service.get_by_name(user_id, name);
        let (setting_time_str, is_latest_change) = match &db_setting {
            None => {(None, false)}
            Some(db_setting) => {(Some(db_setting.value.clone()), true)}
        };

        let mut date_time = None;
        let mut size = 0;
        let now = Local::now();
        if is_latest_change {
            date_time = parse_to_time_by_format(&setting_time_str.unwrap(), "%Y-%m-%d %H:%M");
            size = do_business(date_time, is_latest_change).await;

            let mut im_setting = db_setting.unwrap();
            im_setting.value = format_time_by_format(&now, "%Y-%m-%d %H:%M");
            let setting_service = ImSettingService;
            setting_service.update(&mut im_setting);

        } else {
            let setting_time = now;
            size = do_business(date_time, is_latest_change).await;

            date_time = Some(setting_time.add(TimeDelta::minutes(-1)));
            let mut im_setting = ImSetting {
                id: None,
                name: String::from(name),
                current_user_id: user_id.clone(),
                description: None,
                value: format_time_by_format(&date_time.unwrap(), "%Y-%m-%d %H:%M"),
                ext1: None,
                ext2: None,
                create_time: Local::now(),
                update_time: Local::now(),
            };
            let setting_service = ImSettingService;
            setting_service.insert(&mut im_setting);
        }
        size
    }
}